"""
Module to support Bank Operations in distributed way using the Passive
Replication model.

BankCoordinator is a Bank Interface that also handles passive
replication by listening for slaves on a desired port. Slaves should
connect to this coordinator to join his group.

Last Modified: $Id: coordinator.py 45 2005-06-16 14:20:13Z dafny $

@see: L{bank.Bank}, L{bank.VersionedBank}

@author: U{Gustavo Sverzut Barbieri <barbieri@gmail.com>}
@license: GNU GPL
@copyright: (C) 2005 Gustavo Sverzut Barbieri
@version: $Rev: 45 $
"""

import os
import sys
import time
import socket
import commands
import tempfile
import threading

from threading import Thread
from bank import VersionedBank
from commandserver import CommandRequestHandler
from debug import debug, format_exc

from Queue import Queue, Empty
from SocketServer import ThreadingTCPServer, StreamRequestHandler


class Request( object ):
    """Represent a request from Client to Coordinator.

    A request may be put in a RequestQueue and after it's completed, the
    Coordinator must call C{request.completed( some_value )}, where
    C{some_value} can be the result of request evaluation.

    The Client should wait for request completion with C{request.wait()},
    which blocks until the request is processed and completed.

    @todo: doctests!
    """
    def __init__( self, command, arguments ):
        """Creates a new Request.

        @param command: command to be executed.
        @param arguments: arguments to command.
        """
        self.response  = Queue( 1 )
        self.command   = command
        self.arguments = arguments
    # __init__()


    def wait( self ):
        """Blocks until request is processed and completed, returning it's
        results.

        @return: what ever value was passed to C{completed()}.
        """
        debug( 3, "Wait for request %s" % self )
        return self.response.get()
    # wait()
    result = wait


    def completed( self, value=None ):
        """Notify the request was processed and completed.

        @param value: some value that should be passed to who's waiting this
                      request with C{wait()}.
        """
        debug( 3, "Completed request %s with value=%s" % ( self, value ) )
        self.response.put( value )
    # completed()


    def __str__( self ):
        attributes = [
            'id="%s"' % ( id( self ), ),
            'command="%s"' % ( self.command, ),
            'arguments="%s"' % ( self.arguments, ),
            ]
        l = self.__dict__.keys()
        l.sort()
        for attrname in l:
            attrval = getattr( self, attrname )
            if not callable( attrval ) and \
                   attrname not in ( "response", "command", "arguments" ):
                attributes.append( '%s="%s"' % ( attrname, attrval ) )

        return '[Request %s/]' % " ".join( attributes )
    # __str__()
# Request





class ClientRequest( Request ):
    """Requests from Clients.

    Clients should put this kind of request in the request queue.
    """
    def __init__( self, id, command, arguments ):
        """Creates a new Request.

        @param id: request identifier, must be unique.
        @param command: command to be executed.
        @param arguments: arguments to command.
        """
        Request.__init__( self, command, arguments )
        self.id = id
    # __init__()
# ClientRequest





class SlaveRequest( Request ):
    """Requests for Slaves.

    Coordinator will put this kind of request to Slaves process.
    """
    def __init__( self, thread, command, arguments ):
        """Creates a new Request.

        @param thread: slave thread.
        @param command: command to be executed.
        @param arguments: arguments to command.
        """
        Request.__init__( self, command, arguments )
        self.thread = thread
    # __init__()
# SlaveRequest





class RequestQueue( Queue ):
    """Multi-producer, Multi-Consumer request queue.

    @note: requests should be instance of L{Request}.
    """
    def __init__( self , maxsize=0 ):
        Queue.__init__( self, maxsize )
    # __init__()


    def __str__( self ):
        s = [ "%s:" % self.__class__.__name__ ]
        if len( self.queue ) < 1:
            s.append( "   <Empty>" )
        else:
            for e in self.queue:
                s.append( "   %s" % e )
        return "\n".join( s )
    # __str__()


    def put( self, request, block=True, timeout=None ):
        """Put a request in this queue.

        @param request: a L{Request} instance that should enter the queue.
        @type  request: L{Request}

        @param block: if this call should block if the queue is full.
        @type  block: L{bool}

        @param timeout: how many seconds this thread should block until an
                        L{Queue.Full} exception is raised. If None, it will
                        block until there is room for request in the queue.
        """
        if not isinstance( request, Request ):
            raise TypeError( "request must be an instance of Request" )

        debug( 3, "Enqueue request: %s to %r" % ( request, self ) )
        Queue.put( self, request, block, timeout )
    # put()
    add = put
# RequestQueue






class Coordinator( Thread ):
    """Passive Replication Coordinator.

    This thread handle the request queue, process requests and then send
    updates to slaves.

    Slaves are handled by internal slave server at address specified by
    slave_server_address. They should interpret command as the commandserver
    does.
    """
    def __init__( self, slave_server_address, slave_timeout=None,
                  versionedbank=None, requests_log=None, req_ver_map=None ):
        """Create a new Coordinator instance.

        This coordinator will wait for slaves at C{slave_server_address}.

        @param slave_server_address: where to wait for slaves, must be a tuple
                                     C{( address, port )}, ie
                                     C{( '', 8000 )}.
        @type  slave_server_address: L{tuple}

        @param slave_timeout: how many seconds before issuing timeout on
                              network operations.
        @type  slave_timeout: L{float}

        @param versionedbank: bank to use. If None, will create a new one.
        @type  versionedbank: L{VersionedBank}

        @param requests_log: log with already done requests.
        @type  requests_log: L{dict}

        @param req_ver_map: log with maps from requests and db versions.
        @param req_ver_map: L{dict}
        """

        Thread.__init__( self, name='Coordinator',
                         target=self.action )

        self.keep_running = False

        self.slave_threads = []
        self.request_queue = RequestQueue()

        if versionedbank is None:
            self.__bank = VersionedBank()
        else:
            if isinstance( versionedbank, VersionedBank ):
                self.__bank = versionedbank
            else:
                raise TypeError( "versionedbank should be an instance of " \
                                 "bank.VersionedBank" )

        if requests_log is None:
            self.__requests = {}
        else:
            self.__requests = requests_log

        if req_ver_map is None:
            self.__req_ver = {} # Maps requests to db version
        else:
            self.__req_ver = req_ver_map

        if slave_timeout is None:
            slave_timeout = 5.0

        self.slave_server = Coordinator.SlaveServer( slave_server_address,
                                                     self, slave_timeout )
    # __init__()

    def __str__( self ):
        s = [
            ">>> Bank Status:",
            str( self.__bank ),
            "",
            ">>> Requests:",
            str( self.__requests ),
            "",
            ">>> Requests and their DB Version association:",
            str( self.__req_ver ),
            "",
            ">>> Request Queue:",
            str( self.request_queue ),
            "",
            ">>> Slave Threads:" ]
        for t in self.slave_threads:
            s.append( "   * %s" % t )

        return "\n".join( s )
    # __str__()


    def start( self ):
        self.keep_running = True
        Thread.start( self )
        self.slave_server.start()
    # start()


    def stop( self ):
        self.keep_running = False
        self.slave_server.stop()
    # stop()


    def dump( self ):
        fname = tempfile.mktemp( ".dump",
                                 "%s-" % os.path.basename( sys.argv[ 0 ] ) )
        f = open( fname, "w" )
        f.write( str( self.__bank ) )

        f.write( '\nRequests:\n' )
        reqs = self.__requests.items()
        reqs.sort( lambda a, b: cmp( a[ 0 ], b[ 0 ] ) )
        f.write( str( reqs ) )

        f.write( '\n\nRequests and associated versions:\n' )
        reqver = self.__req_ver.items()
        reqver.sort( lambda a, b: cmp( a[ 0 ], b[ 0 ] ) )
        f.write( str( reqver ) )
        f.write( '\n' )
        f.close()

        debug( 0, "Status dumped to file: %s" % fname )
        return fname
    # dump()


    def do_dump( self ):
        v = self.dump()
        dumps = [ ]
        requests = []
        # Request dump from slaves
        for t in self.slave_threads:
            req = SlaveRequest( t, commands.DUMP, ( None, ) )
            requests.append( req )
            t.request_queue.put( req )

        # Wait for slave replies
        for req in requests:
            ret = req.wait()
            if not ret:
                debug( 1, "Slave failed: %s" % req.thread )
            else:
                ca = req.thread.client_address
                dumps.append( "remote dump %s:%s: %s" % \
                              ( ca[ 0 ], ca[ 1 ], ret ) )
        dumps.append( "local dump: %s" % v )
        return "\n".join( dumps )
    # do_dump()


    def action( self ):
        """Waits for items in the request queue, process them and deliver data.

        Clients should put L{Request}s in the L{RequestQueue} C{request_queue}.
        """
        request_queue = self.request_queue
        while self.keep_running:
            try:
                request = request_queue.get( True, 1 )
            except Empty:
                # timeout is to handle server stopping
                continue

            debug( 1, "Handle: %s" % request )

            if request.command == "dump":
                v = self.do_dump()
                request.completed( v )
                continue


            if request.id in self.__requests:
                # Request already executed
                v = self.__requests[ request.id ]
                request.completed( v )

            else:
                # Execute request and remember it

                version = self.__bank.get_db_version()
                f = getattr( self.__bank, request.command )
                v = f( * request.arguments )
                self.__requests[ request.id ] = v
                self.__req_ver[ version ]     = request.id

                # Propagate data to slaves
                data = self.get_update_data( version )

                if data and ( data[ 0 ] or data[ 1 ] ):
                    # Send data to threads
                    requests = []
                    for t in self.slave_threads:
                        req = SlaveRequest( t, commands.UPDATE_DATA,
                                            ( data, ) )
                        requests.append( req )
                        t.request_queue.put( req )

                    # Wait for thread responses
                    for req in requests:
                        ret = req.wait()
                        if ret == False:
                            debug( 1, "Slave failed: %s" % req.thread )

                request.completed( v )
            # end if request.id in self.__requests

            debug( 2, "coordinator is free to process next request" )

        # end while
        debug( 1, "%s stopped." % self.__class__.__name__ )
    # action()


    def get_update_data( self, version1=0, version2=None ):
        """Get updated data between two versions.

        Data is a tuple::

           ( requests, journal entries, data base changes )

        where:
         - B{requests:} is a list of tuples::
              ( request identifier, version, request value )
         - B{journal entries:} list of data base journal entries.
           May replay this journal to get the updated data.

        @return: what changed between versions.
        @rtype: L{tuple}

        @see: L{VersionedBank.db_journal_entries}
        @see: L{VersionedBank.db_diff}
        """
	if version2 is None:
	    version2 = self.__bank.get_db_version()

	if version1 > version2:
	    raise ValueError( "version1 must be less than version2" )

        re = []
        for i in xrange( version1, version2 ):
            rid = self.__req_ver[ i ]
            re.append( ( rid, i, self.__requests[ rid ] ) )

        # TODO: we need to send the base version if we use shrink_journal!
        je = self.__bank.db_journal_entries( version1, version2 )

        return ( re, je )
    # get_update_data()





    class SlaveServer( ThreadingTCPServer ):
        """Slave Server: handle requests from slaves.

        This TCP server uses threads to handle incoming requests from slaves.
        Each slave thread wait for commands on its own C{request_queue}.
        These queues are one-sized, so slaves or coordinators should block
        until the other side is processing.
        """

        allow_reuse_address = True
        def __init__( self, server_address, coordinator, timeout=None ):
            self.coordinator = coordinator
            SlaveServer = Coordinator.SlaveServer
            ThreadingTCPServer.__init__( self, server_address,
                                         SlaveServer.SlaveRequestHandler )
            self.run = False
            self.timeout = timeout
            self.thread = Thread( target=self.serve_forever )
	    self.thread.setName( "Slave Server" )
        # __init__()


        def start( self ):
            """Start server.

            @note: This launches a new thread, thus is non-blocking.
            """
            debug( 1, "Server %s started. Listening to %s:%s" % \
                   ( self.__class__.__name__, self.server_address[ 0 ],
                     self.server_address[ 1 ] ) )
            self.run = True
            self.thread.start()
        # start()


        def stop( self ):
            """Stop server."""
            self.run = False
            self.thread.join()
            self.server_close()
            debug( 1, "Server %s stopped." % self.__class__.__name__ )
        # stop()


        def serve_forever( self ):
            """Handle requests."""
            while self.run:
                self.handle_request()
            debug( 1, "User required %s to stop." % self.__class__.__name__ )
        # serve_forever()


        def get_request( self ):
            # Reimplemented to avoid blocking forever
            pto = self.socket.gettimeout()
            self.socket.settimeout( 1 )
            try:
                r = self.socket.accept()
                self.socket.settimeout( pto )
                return r
            except socket.timeout:
                self.socket.settimeout( pto )
                raise socket.error( "No connections within timeout, " \
                                    "try again." )
        # get_request()

	def process_request( self, request, client_address ):
		# Overload to change thread's name
		self.process_request_lock.acquire()
		self.process_request_number += 1
		tid = self.process_request_number
		self.process_request_lock.release()

		t = threading.Thread( target = self.process_request_thread,
				      args = (request, client_address),
				      name = "Slave-%d" % tid )
		if self.daemon_threads:
			t.setDaemon( 1 )

		t.start()
	# process_request()
	process_request_number = 0
	process_request_lock   = threading.RLock()


        class SlaveRequestHandler( CommandRequestHandler ):
            """Request handler for incoming slave connections.

            This will handle incoming connections, handle commands, put them in
            the request queue and then reply to other peer.

            Methods that begin with 'do_slave_' are to handle slave commands,
            those with 'do_coordinator_' are to handle coordinator commands.
            These functions should return True if they worked or False if
            the Slave failed, thus terminating this instance.

            Coordinator can send requests using C{request_queue}.
            """


            def setup( self ):
                """Setup a new connection.

                This put this slave thread on the Coordinator's
                C{slave_threads} list.
                """
                CommandRequestHandler.setup( self )

                self.slave_commands = {
                    commands.RETRIEVE_DB: self.do_slave_retrieve_db,
                    }

                self.coordinator_commands = {
                    commands.UPDATE_DATA: self.do_coordinator_update_data,
                    commands.DUMP: self.do_coordinator_dump,
                    }

                self.run = True
                self.request_queue = RequestQueue( 1 )

                self.server.coordinator.slave_threads.append( self )

                self._debug( 1, "Got connection." )
            # setup()


            def finish( self ):
                """Ends a connection.

                This takes this slave thread from the Coordinator's
                C{slave_threads} list.
                """
                CommandRequestHandler.finish( self )
                self.server.coordinator.slave_threads.remove( self )

                self._debug( 1, "End connection." )
            # finish()


            def send_quit( self ):
                """Have the other peer to quit."""
                self.send( commands.QUIT, None )
            # send_quit()


            def do_slave_retrieve_db( self, data ):
                """Process slave's RETRIEVE_DB command, thus send the whole
                data base to it.
                """
                self._debug( 2, "Sending whole database to client" )
                data = self.server.coordinator.get_update_data()
                self.send( commands.UPDATE_DATA, data )
                return True
            # do_retrieve_db()


            def do_coordinator_update_data( self, data ):
                """Process coordinator's UPDATE_DATA command.
                """
                self._debug( 2, "Coordinator asked to update slave's data." )
                self.send( commands.UPDATE_DATA, data )
                return True
            # do_coordinator_update_data()


            def do_coordinator_dump( self, data ):
                self._debug( 2, "Coordinator asked slave to dump status." )
                self.send( commands.DUMP, None )
                cmd, data = self.recv()
                if cmd == commands.ACK:
                    return data
                else:
                    return None
            # do_coordinator_dump()


            def handle( self ):
                """Process commands from slave.
                """
                try:
                    self.request.setblocking( 1 )
                    self.request.settimeout( self.server.timeout )
                    while self.server.run:
                        try:
                            request = self.request_queue.get_nowait()
                            # Process Coordinator requests

                            self._debug( 3, "Handle: %s" % request )

                            cmd  = request.command
                            args = request.arguments
                            try:
                                f = self.coordinator_commands[ cmd ]
                                v = f( * args )

                                self.send_ack()
                                if not self.recv_ack():
                                    self._debug( 0, "Client should send ACK!" )
                                    self.run = False
                                    self.send_quit()
                                    request.completed( False )
                                else:
                                    request.completed( v )

                            except KeyError, e:
                                self._debug( 0,
                                             ( 'Coordinator insanely ' \
                                               'requested the inexistent ' \
                                               'command "%s"' ) % cmd )
                                request.completed( False )

                            except socket.timeout, e:
                                self._debug( 0,
                                             """\
------------------------------------------------------------
Socket operation timed out in %(timeout)s seconds.
Exception: %(exception)s
Traceback:

%(traceback)s

------------------------------------------------------------
""" % { 'timeout': self.server.timeout,
        'exception': e,
        'traceback': format_exc() } )
                                self.send_quit()
                                self.run = False
                                request.completed( False )

                            except socket.error, e:
                                self._debug( 0,"Lost connection: %s" %  e )
                                request.completed( False )

                            continue

                        except Empty:
                            # Process Slave requests
                            if not self.run:
                                break

                            self.request.setblocking( 0 )
                            try:
                                cmd, args = self.recv()
                                if cmd is None:
                                    return # Stop!

                                self.request.setblocking( 1 )
                                self.request.settimeout( self.server.timeout )
                                f = self.slave_commands[ cmd ]
                                f( args )

                                self.send_ack()
                                if not self.recv_ack():
                                    self._debug( 0, "Client should send ACK!" )
                                    self.run = False
                                    self.send_quit()

                                continue

                            except KeyError, e:
                                self._debug( 0,
                                             "Command '%s' doesn't exists!" % \
                                             cmd )

                            except socket.timeout, e:
                                self._debug( 0,
                                             """\
------------------------------------------------------------
Socket operation timed out in %(timeout)s seconds.
Exception: %(exception)s
Traceback:

%(traceback)s

------------------------------------------------------------
""" % { 'timeout': self.server.timeout,
        'exception': e,
        'traceback': format_exc() } )
                                self.send_quit()
                                self.run = False

                            except socket.error, e:
                                # No data on socket...
                                self.request.setblocking( 1 )
                                self.request.settimeout( self.server.timeout )
                                time.sleep( 0.0001 ) # Avoid 100% cpu

                except socket.error, e:
                    self._debug( 0, "Lost connection: %s" % e )
            # handle()


            def _debug( self, level, message ):
                """Print message to standard error based on current debug level and
                message level.

                Message is only printed if C{level >= debug.level}.

                @param level: message level
                @type  level: L{int}
                @param message: what to print.
                """
                debug( level, "%s: %s:%s: %s" % \
                       ( self.__class__.__name__,
                         self.client_address[ 0 ],
                         self.client_address[ 1 ],
                         message ) )
            # _debug()
        # SlaveRequestHandler
    # SlaveServer
# Coordinator
